package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.dto.UpdateArticleMessage;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.StringUtils;

import java.time.Duration;

@EnableBinding(value = ArticleProcess.class)
public class ArticleListener {

    @Value("${commit.time}")
    private String commitTime;

    //指定数据来源的topic
    @StreamListener(value = "article_behavior")
    @SendTo(value = "article_result")
    public KStream<String,String> process(KStream<String,String> input){
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //首先获取文章id
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                Long articleId = updateArticleMessage.getArticleId();
                //构建返回的键值对
                return new KeyValue<>(articleId.toString(),value);
            }
        });

        //分组 -- 根据文章id进行分组
        KGroupedStream<String, String> groupByKey = map.groupByKey();
        //统计时间窗口的数据
        TimeWindowedKStream<String, String> windowedBy = groupByKey.windowedBy(TimeWindows.of(Duration.ofMillis(Long.parseLong(commitTime))));
        //进行聚合处理
        //初始聚合
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                //在同一个时间窗口内第一次接收到消息，返回聚合结果
                System.out.println("开始进行聚合。。。。。。。");
                return null;
            }
        };

        //聚合处理
        Aggregator<String, String, String> aggregator = new Aggregator<String, String, String>() {
            //key 是沙漠后面定义的文章id   value则是updateArticleMessage aggregate是在同一个时间窗口内上一次聚合的结果
            @Override
            public String apply(String key, String value, String aggregate) {
                ArticleStreamMessage message = null;
                //判断上一次聚合是否为空  为空则新增
                if (StringUtils.isEmpty(aggregate)) {
                    message = new ArticleStreamMessage();
                    message.setArticleId(Long.parseLong(key));
                } else {
                    //不为空  结果直接从上一次提取
                    message = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }
                //提取本次接收到的消息
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                //处理本次消息
                switch (updateArticleMessage.getType()) {
                    //type分别为： 0 阅读  1点赞  2评论   3收藏
                    case 0:
                        message.setView(message.getView() + updateArticleMessage.getAdd());
                        break;
                    case 1:
                        message.setLike(message.getLike() + updateArticleMessage.getAdd());
                        break;
                    case 2:
                        message.setComment(message.getComment() + updateArticleMessage.getAdd());
                        break;
                    case 3:
                        message.setCollect(message.getCollect() + updateArticleMessage.getAdd());
                        break;
                }
                //将本次更新的结果保存到聚合的中间结果
                String json = JSON.toJSONString(message);
                return json;
            }
        };

        //获取聚合结果
        KTable<Windowed<String>, String> aggregate = windowedBy.aggregate(init, aggregator);
        //转换聚合结果
        KStream<String, String> stream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });

        return stream;
    }
}
